package org.ala.tiktools.rocketmq.serializable.kryo;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.ala.tiktools.rocketmq.MQMessage;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.concurrent.Callable;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * kryo工具类
 *
 * @author ala
 * @date 2024-11-29 09:29
 */
public class KryoTool {


    /**
     *  对象池
     */
    protected GenericObjectPool<Kryo> pool;

    /**
     *  单例
     */
    private KryoTool() {
        init();
    }
    private static KryoTool instance;
    public static KryoTool getInstance() {
        if (instance == null) {
            synchronized (KryoTool.class) {
                if (instance == null) {
                    instance = new KryoTool();
                }
            }
        }
        return instance;
    }

    /**
     *  初始化kryo池子
     */
    protected void init() {
        GenericObjectPoolConfig<Kryo> poolConfig = new GenericObjectPoolConfig<>();
        poolConfig.setMaxTotal(10);
        poolConfig.setMaxIdle(4);
        poolConfig.setMinIdle(1);
        //  借出、还回、空闲时，均检测对象有效性
        poolConfig.setTestOnBorrow(true);
        poolConfig.setTestOnReturn(true);
        poolConfig.setTestWhileIdle(true);

        poolConfig.setNumTestsPerEvictionRun(3);
        poolConfig.setBlockWhenExhausted(false);
        poolConfig.setJmxEnabled(true);

        PooledObjectFactory<Kryo> factory = new BasePooledObjectFactory<Kryo>() {
            @Override
            public Kryo create() throws Exception {
                Kryo kryo = new Kryo();
                kryo.setRegistrationRequired(false);
                kryo.setReferences(true);
                return kryo;
            }
            @Override
            public PooledObject<Kryo> wrap(Kryo obj) {
                return new DefaultPooledObject<Kryo>(obj);
            }
        };

        pool = new GenericObjectPool<>(factory, poolConfig);
    }


    /**
     *  反序列化对象
     */
    public <T> T deserialize(byte[] bytes, Class<T> clazz) throws Exception {
        return withKryo((kryo) -> {
            ByteArrayInputStream inStream = new ByteArrayInputStream(bytes);
            Input in = new Input(inStream,1024);
            try {
                return kryo.readObject(in, clazz);
            } finally {
                in.close();
            }
        });
    }
    /**
     *  序列化对象
     */
    public byte[] serialize(Object objs) throws Exception {
        return withKryo(kryo -> {
            ByteArrayOutputStream outStream = new ByteArrayOutputStream();
            Output out = new Output(outStream);
            try {
                kryo.writeObject(out, objs);
                out.flush();
                return outStream.toByteArray();
            } finally {
                out.close();
            }
        });
    }


    /**
     *  在kryo的作用域内工作
     */
    public <R> R withKryo(Function<Kryo, R> fun) throws Exception {
        //  尝试借出
        Kryo kryo = pool.borrowObject();
        try {
            return fun.apply(kryo);
        } finally {
            pool.returnObject(kryo);
        }
    }



}
